package lib ;

import java.io.* ;
import java.util.* ;
// import java.time.* ;
import java.text.* ;
import java.net.* ;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.avro.shaded.org.apache.avro.Schema;
import org.apache.flink.avro.shaded.org.apache.avro.generic.GenericData;
import org.apache.flink.avro.shaded.org.apache.avro.generic.GenericRecord;
import org.apache.flink.avro.shaded.org.apache.avro.generic.GenericContainer;
import org.apache.flink.avro.shaded.org.apache.avro.generic.GenericDatumWriter;
import org.apache.flink.avro.shaded.org.apache.avro.io.Encoder;
import org.apache.flink.avro.shaded.org.apache.avro.io.EncoderFactory;

import com.datastax.driver.core.Cluster ;
import com.datastax.driver.core.Session ;

public class Utils {
    public static Session mkCqlSession(String cassandras) {
        Cluster.Builder cqlSessionBuilder = Cluster.builder() ;
        for(String node: cassandras.split(",")) {
            String[] nodeParts = node.split(":") ;
            String host = nodeParts[0] ;
            int port = Integer.parseInt(nodeParts[1]) ;
            cqlSessionBuilder.addContactPointsWithPorts(new InetSocketAddress(host, port)) ;
        }
        return cqlSessionBuilder.build().connect() ;
    }
    
    public static boolean hasColdStartEvent(String platformName) {
        if (Arrays.asList("android", "ios", "wechat").contains(platformName)) {
            return true ;
        } else {
            return false ;
        }
    }
    public static String parseChinaTs(long ts) {
        Date dt = new Date(ts);
        SimpleDateFormat bjSdf = new SimpleDateFormat("yyyyMMddHHmm");
        bjSdf.setTimeZone(TimeZone.getTimeZone("Asia/Shanghai"));
        return bjSdf.format(dt);
    }
    
    private static byte[] toByteArray(InputStream in) throws Exception {
        ByteArrayOutputStream buffer = new ByteArrayOutputStream();
        int nRead;
        byte[] data = new byte[16384];
        while ((nRead = in.read(data, 0, data.length)) != -1) {
            buffer.write(data, 0, nRead);
        }
        return buffer.toByteArray();
    }
    private static String toText(InputStream in) throws Exception {
            return new String(toByteArray(in)) ;
    }

    public static String slurpFile(String schemaPath) throws Exception {
        InputStream in = new FileInputStream(schemaPath) ;
        return toText(in) ;
    }
    public static Object avroGet(GenericRecord record, String path) {
        String[] parts = path.split("\\.") ;

        Object result = record ;
        for (int i = 0; i < parts.length; i++) {
            result = ((GenericRecord)result).get(parts[i]) ;
            if (result == null) return null ;
        }
        return result ;
    }
    
    public static Object jsonGet(Object json, List<Object> path) throws Exception {
        if (json == null) return null ;
        Object result = json ;
        for(int i = 0; i < path.size(); i++) {
            result = ((Map<String, Object>)result).get(path.get(i)) ;
            if (result == null) return null ;
        }
        return result ;
    }
    public static Object jsonGet(String s, List<Object> path) throws Exception {
        if (s == null) return null ;
        return jsonGet(new ObjectMapper().readValue(s, Object.class), path) ;
    }
    public static Object jsonSet(Object json, List<Object> path, Object val) throws Exception {
        Object target = json ;
        for(int i = 0; i < path.size(); i++) {
            if (i == path.size() - 1) {
                Map<Object, Object> targetMap = (Map<Object, Object>) target ;
                targetMap.put(path.get(i), val) ;
            } else {
                Map<Object, Object> targetMap = (Map<Object, Object>) target ;
                target = targetMap.get(path.get(i)) ;
                if (target == null) targetMap.put(path.get(i), new HashMap<Object,Object>())  ;
                target = targetMap.get(path.get(i)) ;
            }
        }
        return json ;
    }
    public static String jsonSet(String s, List<Object> path, Object val) throws Exception {
        ObjectMapper om = new ObjectMapper() ;
        Object json = (s == null)? new HashMap<Object,Object>() : om.readValue(s, Object.class) ;
        
        return om.writeValueAsString(jsonSet(json, path, val)) ;
    }
    public static Object removeTypeInfo(Object o) {
        if (o instanceof Map) {
            Map<String, Object> newM = new HashMap<String, Object>() ;
            
            List<String> typeKeywords = Arrays.asList("string", "long", "int") ;
            for(Map.Entry<String, Object> entry: ((Map<String, Object>) o).entrySet()) {
                String key = entry.getKey() ;
                if (typeKeywords.contains(key)) {
                    return removeTypeInfo(entry.getValue()) ;
                }
                newM.put(key, removeTypeInfo(entry.getValue())) ;
            }
            return newM ;
        } else if (o instanceof List) {
            List<Object> newL = new ArrayList<Object>() ;
            for(Object e: (List<Object>) o) {
                newL.add(removeTypeInfo(e)) ;
            }
            return newL ;
        } else {
            return o ;
        }
    }

    
    public static Object avroToJson(GenericContainer datum) throws Exception {
        ObjectMapper om = new ObjectMapper() ;
        Schema schema = datum.getSchema() ;
        ByteArrayOutputStream out = new ByteArrayOutputStream() ;
        Encoder jsonEncoder = EncoderFactory.get().jsonEncoder(schema, out);
        GenericDatumWriter<GenericContainer> writer = new GenericDatumWriter<GenericContainer>(schema) ;
        writer.write(datum, jsonEncoder) ;
        jsonEncoder.flush();
        String json = new String(out.toByteArray()) ;
        
        return removeTypeInfo(om.readValue(json, Object.class)) ;
    }
    
    public static String extractPlatformName(GenericRecord record) {
        String result = "unknown" ;
        if (record.get("is_android") != null) {
            result = "android" ;
        } else if (record.get("is_ios") != null) {
            result = "ios" ;
        } else if (record.get("is_wechat") != null) {
            result = "wechat" ;
        } else if (record.get("is_web") != null) {
            result = "web" ;
        } else if (record.get("is_h5") != null) {
            result = "h5" ;
        }
        return result ;
    }
    public static String extractPlatformName(Map<String, Object> record) {
        String result = "unknown" ;
        if (record.get("is_android") != null) {
            result = "android" ;
        } else if (record.get("is_ios") != null) {
            result = "ios" ;
        } else if (record.get("is_wechat") != null) {
            result = "wechat" ;
        } else if (record.get("is_web") != null) {
            result = "web" ;
        } else if (record.get("is_h5") != null) {
            result = "h5" ;
        }
        return result ;
    }
    public static String extractEventName(GenericRecord record) {
        String result = "unknown" ;
        if (record.get("is_cold_start") != null) {
            result = "cold_start" ;
        } else if (record.get("is_hot_start") != null) {
            result = "hot_start" ;
        } else if (record.get("is_use_end") != null) {
            result = "use_end" ;
        } else if (record.get("is_view_page") != null) {
            result = "view_page" ;
        } else if (record.get("is_click") != null) {
            result = "click" ;
        } else if (record.get("is_custom") != null) {
            result = "custom" ;
        }
        return result ;
    }
    public static String extractEventName(Map<String, Object> record) {
        String result = "unknown" ;
        if (record.get("is_cold_start") != null) {
            result = "cold_start" ;
        } else if (record.get("is_hot_start") != null) {
            result = "hot_start" ;
        } else if (record.get("is_use_end") != null) {
            result = "use_end" ;
        } else if (record.get("is_view_page") != null) {
            result = "view_page" ;
        } else if (record.get("is_click") != null) {
            result = "click" ;
        } else if (record.get("is_custom") != null) {
            result = "custom" ;
        }
        return result ;
    }

    /*
    public static String extractPlatformName(String schemaName) {
        String result = "unknown" ;
        if (schemaName.equals("DcSdkAndroid")) {
            result = "android" ;
        } else if (schemaName.equals("DcSdkIos")) {
            result = "ios" ;
        } else if (schemaName.equals("DcSdkWeb")) {
            result = "web" ;
        } else if (schemaName.equals("DcSdkWechat")) {
            result = "wechat" ;
        } else if (schemaName.equals("DcSdkH5")) {
            result = "h5" ;
        }
        return result ;
    }

    public static String extractEventName(String schemaName) {
        String result = "unkown" ;
        if (schemaName.startsWith("EventColdStart")) {
            result = "cold_start" ;
        } else if (schemaName.startsWith("EventHotStart")) {
            result = "hot_start" ;
        } else if (schemaName.startsWith("EventUseEnd")) {
            result = "use_end" ;
        } else if (schemaName.startsWith("EventViewPage")) {
            result = "view_page" ;
        } else if (schemaName.startsWith("EventClick")) {
            result = "click" ;
        } else if (schemaName.startsWith("EventCustom")) {
            result = "custom" ;
        }
        return result ;
    }
    */
}
